package com.ai.bitcoin.queue;

import com.ai.bitcoin.exception.ThreadShutDownException;
import com.ai.bitcoin.service.helper.NsqServiceHelper;
import com.ai.bitcoin.util.JacksonUtils;
import com.ai.bitcoin.vo.Anno2BccVo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Callable;

public class AnnoThread  implements Callable<Void> {

    private static Logger logger = LoggerFactory
            .getLogger(AnnoThread.class);

    private MessageQueue annoQueue;

    private NsqServiceHelper nsqServiceHelper;

    public AnnoThread(MessageQueue annoQueue,NsqServiceHelper nsqServiceHelper) {
        super();
        this.annoQueue = annoQueue;
        this.nsqServiceHelper = nsqServiceHelper;
    }

    @Override
    public Void call() throws Exception {
        logger.warn("announce Thread starting ...");
        while (true) {
            try {
                String data = this.annoQueue.receive();
                Anno2BccVo anno2BccVo = JacksonUtils.fromJson(data, Anno2BccVo.class);
                nsqServiceHelper.distributorAnno(anno2BccVo);
            } catch (ThreadShutDownException e) {
                if(logger.isDebugEnabled()) {
                    logger.debug("message thread shutdown");
                }
                break;
            } catch (Exception e1) {
                logger.error(e1.getMessage(), e1);
            }
        }
        return null;
    }

}
